<?php
namespace lactone\Core\Queue\Driver;

class MysqlDriver implements QueueDriverInterface{

    static $Connects;

    public  $pdo;
    public  $key;

    /**
     * @var $Connect \PDO
     */
    private $Connect;

    private $table = "queue_mysql";

    const  NEW_STATUS = 0;
    const  USED_STATUS = 1;

    public function push($data){
        $this->initPdo();

        $this->Connect->beginTransaction();
        try{
            $stmt = $this->Connect->prepare(" Insert into ".$this->table." (`key` , `value` ,`status` ,`create_at`) VALUES (:key,:value,".self::NEW_STATUS.",:created_at) ");
            $stmt->bindParam(":key", $this->key);
            $stmt->bindParam(":value",$data,\PDO::PARAM_STR);
            $stmt->bindValue(":created_at",time());
            $stmt->execute();
        }catch (\Exception $e){
            $this->Connect->rollBack();
            return false;
        }

        $this->Connect->commit();


        return true;
    }

    /**
     * @return bool
     */
    public function shift(){
        $this->initPdo();

        $data = false;
        $this->Connect->beginTransaction();
        $sql = "select `value`,id from ".$this->table." where `key` = '".$this->key."'  and status = ".self::NEW_STATUS." limit 1 for update ";
        $rs  = $this->Connect->query($sql);
        if( ( $col = $rs->fetchAll() ) && (count($col) > 0 ) ){
            $result = $col[0];
            $data = $result['value'];
            //$this->Connect->exec(" update ".$this->table." set status = ".self::USED_STATUS ." where id = ".$result['id']);

            $this->Connect->exec(" delete from ".$this->table." where id =  ".$result['id']);
        }

        $this->Connect->commit();


        return $data;

    }

    /**
     * 是否为空
     */
    public function isEmpty(){
        $this->initPdo();

        $sql = "select count(*) from ".$this->table." where `key` = '".$this->key."'  and status = ".self::NEW_STATUS;

        //\Lactone::$log->debug($sql);

        $col = 0;
        if($rs  = $this->Connect->query($sql)){
            $col = $rs->fetchColumn();
        }else{
            $this->logError();
        }


        if($col > 0 ){
            return false;
        }
        return true;
    }

    /**
     * 数据量总数
     */
    public function count(){
        $this->initPdo();

        $sql = "select count(*) from ".$this->table." where `key` = '".$this->key."'  and status = ".self::NEW_STATUS;

        $col = 0;
        if($rs  = $this->Connect->query($sql)){
            $col = $rs->fetchColumn();
        }else{
            $this->logError();
        }


        if($col > 0 ){
            return $col;
        }

        return 0;

    }

    public function initConfig($Configs){
        if(!isset($Configs['pdo'])){
            throw new \Exception("need set Pdo Key");
        }

        foreach($Configs as $key => $value){
            $this->$key = $value;
        }

        return true;
    }

    public function reset(){
        $pdo = array_merge(['options'=>[]],$this->pdo);

        $key = md5(json_encode($pdo));

        if(isset(self::$Connects[$key])){
            unset(self::$Connects[$key]);
        }

        return true;
    }

    /**
     * 链接pdo的信息
     */
    protected function initPdo(){
        $pdo = array_merge(['options'=>[]],$this->pdo);

        $key = md5(json_encode($pdo));

        if(isset(self::$Connects[$key])){

            try {
                self::$Connects[$key]->query('SELECT 1');
            } catch (PDOException $e) {
                self::$Connects[$key] = NULL;
                self::$Connects[$key] = new \PDO($pdo['dsn'],$pdo['username'],$pdo['password'],$pdo['options']);
            }

        }else{
            self::$Connects[$key] = new \PDO($pdo['dsn'],$pdo['username'],$pdo['password'],$pdo['options']);
        }

        $this->Connect = self::$Connects[$key];
    }

    protected function closePdo(){
        if($this->Connect){
            $this->Connect = NULL;
        }
    }

    protected function logError(){
        \Lactone::notice("[Queue MsqlDriver] ".$this->Connect->error ." ".$this->Connect->errorInfo);
    }
}
?>